package cn.azzhu.day04
import cn.azzhu.day02.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
/**
 * ProcessFunction将数据发送到侧输出流
 * @author azzhu
 * @create 2020-09-21 18:31:32
 */
object FreezingAlarm {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.addSource(new SensorSource)
      .process(new FreezingAlarmFunction)

    //打印常规输出
   // stream.print()

    //打印侧输出,侧输出标签的名字必须一样
    stream.getSideOutput(new OutputTag[String]("freezing-alarm")).print()

    env.execute("FreezingAlarm")
  }

  // `ProcessFunction`处理的是没有keyBy的流
  class FreezingAlarmFunction extends ProcessFunction[SensorReading,SensorReading] {
    //定义一个侧输出标签，实际上就是侧输出流的名字
    lazy val freezingAlarmOut = new OutputTag[String]("freezing-alarm")
    override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
      if(value.temperature < 32.0) {
        //第一个参数是侧输出标签，第二个参数是发送的数据
        ctx.output(freezingAlarmOut,s"${value.id}的传感器低温报警！")
      }
      //将所有读数发送到常规输出
      out.collect(value)
    }
  }
}
